package guda.mrpc.core.transport.tcp;


import guda.mrpc.core.RpcRequest;
import guda.mrpc.core.common.ExecutorServiceFactory;
import guda.mrpc.core.server.Server;
import guda.mrpc.core.transport.Hearbeat;
import guda.mrpc.core.transport.ServerHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;


public class TcpServerHandler extends ServerHandler<Object> {


    private Server server;

    public TcpServerHandler(Server server) {
        this.server = server;
    }


    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        log.warn("channel inactive,remote address:" + getRemoteAddress() + ",local address:" + getLocalAddress());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof RpcRequest){
            RpcRequest request=(RpcRequest)msg;
            TcpResponseCallback tcpResponseCallback = new TcpResponseCallback(ctx, request, server.getServiceProviderContainer());
            ExecutorServiceFactory.getExecuteService().submit(tcpResponseCallback);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Tcp server handler error", cause);
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.WRITER_IDLE){
                ctx.writeAndFlush(Hearbeat.NullHearbeat.instance);
            }
        }else{
            super.userEventTriggered(ctx, evt);
        }
    }
}
